package ai.people.netmon.msgsearch.service.impl;

import ai.people.core.common.properties.elastic.ElasticSocialAccountIndex;
import ai.people.core.common.util.ApplicationContextUtil;
import ai.people.netmon.framework.domain.jx.vo.SocialAccountCardReqVO;
import ai.people.netmon.framework.model.response.PageResponse;
import ai.people.netmon.msgsearch.domain.entity.SocialAccountCard;
import ai.people.netmon.msgsearch.service.IJxSocialAccountCardService;
import com.google.common.collect.Lists;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.MatchQueryBuilder;
import org.elasticsearch.index.query.MultiMatchQueryBuilder;
import org.elasticsearch.index.query.Operator;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.data.elasticsearch.core.SearchHit;
import org.springframework.data.elasticsearch.core.SearchHits;
import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
import org.springframework.data.elasticsearch.core.query.NativeSearchQuery;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAdder;
import java.util.stream.Collectors;

/**
 * @author LiGuanNan
 * @date 2022/6/30 17:52
 */
@Service
public class JxSocialAccountCardServiceImpl implements IJxSocialAccountCardService {

    @Autowired
    @Qualifier("flElasticsearchRestTemplate")
    private ElasticsearchRestTemplate flElasticsearchRestTemplate;

    @Autowired
    @Qualifier("elasticsearchRestTemplate")
    private ElasticsearchRestTemplate elasticsearchRestTemplate;

    @Autowired
    private ElasticSocialAccountIndex elasticSocialAccountIndex;

    private static final String AFRICA_COUNTRY_NAMES = "\\\"Burundi\\\" \\\"Botswana\\\" \\\"Burkina Faso\\\" \\\"Congo\\\" \\\"Côte d'Ivoire\\\" \\\"Benin\\\" \\\"Cameroon\\\" \\\"Central African Rep.\\\" \\\"Dem. Rep. Congo\\\" \\\"Angola\\\" \\\"Comoros\\\" \\\"Djibouti\\\" \\\"Cape Verde\\\" \\\"Gabon\\\" \\\"Algeria\\\" \\\"Ethiopia\\\" \\\"Guinea\\\" \\\"Ghana\\\" \\\"Eritrea\\\" \\\"Gambia\\\" \\\"Egypt\\\" \\\"Guinea-Bissau\\\" \\\"Eq. Guinea\\\" \\\"Liberia\\\" \\\"Kenya\\\" \\\"Lesotho\\\" \\\"Libya\\\" \\\"Morocco\\\" \\\"Mali\\\" \\\"Madagascar\\\" \\\"Mozambique\\\" \\\"Malawi\\\" \\\"Mauritania\\\" \\\"Namibia\\\" \\\"Niger\\\" \\\"Rwanda\\\" \\\"W. Sahara\\\" \\\"Nigeria\\\" \\\"Sudan\\\" \\\"Senegal\\\" \\\"S. Sudan\\\" \\\"Sierra Leone\\\" \\\"Somaliland\\\" \\\"Somalia\\\" \\\"Swaziland\\\" \\\"São Tomé and Principe\\\" \\\"Tunisia\\\" \\\"Togo\\\" \\\"Chad\\\" \\\"Tanzania\\\" \\\"Zambia\\\" \\\"Uganda\\\" \\\"South Africa\\\" \\\"Zimbabwe\\\"";
    private static final List<String> AVAILABLE_AFRICA_COUNTRY_NAMES = Lists.newArrayList("Burundi", "Botswana", "Cameroon", "Angola", "Algeria", "Ghana", "Egypt");

    private static final String[] FL_SEARCH_FIELDS = new String[]{"uid", "email", "id", "fullName", "firstName", "lastName", "linkedinUsername"};

    private static final String[] YRT_SEARCH_FIELDS = new String[]{"nick_name", "user_name", "id", "urls"};

    @Override
    public PageResponse<SocialAccountCard> searchPage(SocialAccountCardReqVO searchParam) {
        ExecutorService executorService = ApplicationContextUtil.getBean(ExecutorService.class, "defaultThreadPoolExecutor");
        List<Future<PageResponse<SocialAccountCard>>> futures = Lists.newArrayList();
        if(CollectionUtils.isEmpty(searchParam.getDataSourceList())
                || searchParam.getDataSourceList().stream().anyMatch(o-> elasticSocialAccountIndex.getCluster1IndexMap().containsKey(o))){
            Future<PageResponse<SocialAccountCard>> flResponseFuture = executorService.submit(() -> {
                List<String> dataSource = Lists.newArrayList("facebook", "linkedin");
                if (!CollectionUtils.isEmpty(searchParam.getDataSourceList())) {
                    dataSource = searchParam.getDataSourceList();
                }
                return daSearch(searchParam, flElasticsearchRestTemplate, FL_SEARCH_FIELDS, dataSource, 1);
            });
            futures.add(flResponseFuture);
        }
        if(CollectionUtils.isEmpty(searchParam.getDataSourceList())
                || searchParam.getDataSourceList().stream().anyMatch(o-> elasticSocialAccountIndex.getCluster2IndexMap().containsKey(o))){
            Future<PageResponse<SocialAccountCard>> yrtResponseFuture = executorService.submit(() -> {
                List<String> dataSource = Lists.newArrayList("youtube", "reddit", "twitter");
                if (!CollectionUtils.isEmpty(searchParam.getDataSourceList())) {
                    dataSource = searchParam.getDataSourceList();
                }
                return daSearch(searchParam, elasticsearchRestTemplate, YRT_SEARCH_FIELDS, dataSource, 2);
            });
            futures.add(yrtResponseFuture);
        }

        List<SocialAccountCard> list = Lists.newArrayList();
        LongAdder total = new LongAdder();
        AtomicLong maxPages = new AtomicLong(0);
        futures.forEach(f->{
            try {
                PageResponse<SocialAccountCard> pageResponse = f.get();
                list.addAll(pageResponse.getList());
                total.add(pageResponse.getTotal());
                while (true){
                    Long pages = maxPages.get();
                    if(maxPages.compareAndSet(pages, Math.max(pages, pageResponse.getPages()))){
                        break;
                    }
                }
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        });

        return PageResponse.builder(list, total.longValue(), maxPages.get(), searchParam.getOriginPageNo());
    }

    @Override
    public PageResponse<SocialAccountCard> searchPageV2(SocialAccountCardReqVO searchParam) {
        Integer originPageSize = searchParam.getPageSize();
        Map<String, String> cluster1IndexMap = elasticSocialAccountIndex.getCluster1IndexMap();
        Map<String, String> cluster2IndexMap = elasticSocialAccountIndex.getCluster2IndexMap();
        List<String> keyList = new ArrayList<>();
        keyList.addAll(cluster1IndexMap.keySet());
        keyList.addAll(cluster2IndexMap.keySet());

        if(CollectionUtils.isEmpty(keyList)){
            return PageResponse.builderEmpty();
        }

        if(searchParam.getIsAfricaAccount() == 1){
            keyList = keyList.stream().filter(o->o.equalsIgnoreCase("facebook")).collect(Collectors.toList());
        }

        List<Future<PageResponse<SocialAccountCard>>> futures = Lists.newArrayList();
        ExecutorService executorService = ApplicationContextUtil.getBean(ExecutorService.class, "defaultThreadPoolExecutor");
        for(int i = 0; i < keyList.size(); i++){
            String key = keyList.get(i);
            if(originPageSize % keyList.size() != 0 && i == keyList.size() - 1){
                searchParam.setPageSize(originPageSize / keyList.size() + originPageSize % keyList.size() );
            }else {
                searchParam.setPageSize(originPageSize / keyList.size() );
            }
            if(cluster1IndexMap.containsKey(key)){
                futures.add(executorService.submit(() -> daSearch(searchParam, flElasticsearchRestTemplate, FL_SEARCH_FIELDS, Lists.newArrayList(key), 1)));
            }else{
                futures.add(executorService.submit(()-> daSearch(searchParam, elasticsearchRestTemplate, YRT_SEARCH_FIELDS, Lists.newArrayList(key), 2)));
            }
        }

        List<SocialAccountCard> list = Lists.newArrayList();
        LongAdder total = new LongAdder();
        futures.forEach(f->{
            try {
                PageResponse<SocialAccountCard> pageResponse = f.get();
                list.addAll(pageResponse.getList());
                total.add(pageResponse.getTotal());
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        });
        return PageResponse.builder(list, total.longValue(), (total.longValue() + originPageSize - 1) / originPageSize, searchParam.getOriginPageNo());
    }

    public PageResponse<SocialAccountCard> daSearch(SocialAccountCardReqVO searchParam, ElasticsearchRestTemplate restTemplate, String[] searchFields, List<String> dataSource, Integer clusterType) {
        List<SocialAccountCard> socialUsers = Lists.newArrayList();
        long total=0, totalPage=1;

        // 创建查询对象
        NativeSearchQuery query = getNativeSearchQuery(searchParam, searchFields);

        //指定查询索引
        IndexCoordinates indexCoordinates = IndexCoordinates.of(Objects.requireNonNull(elasticSocialAccountIndex.getIndexListBySources(dataSource, clusterType)));;

        // 调用查询接口
        SearchHits<SocialAccountCard> searchHits = null;
        try {
            if(Objects.equals(searchParam.getIsAfricaAccount(), 1) && !Objects.equals(searchParam.getCurrentPage(), searchParam.getOriginPageNo())){
                searchHits = restTemplate.search(getAfricaNativeSearchQuery(searchParam, searchFields), SocialAccountCard.class, indexCoordinates);
            }else {
                searchHits = restTemplate.search(query, SocialAccountCard.class, indexCoordinates);
            }
            // 记录总数
            total = restTemplate.count(query, indexCoordinates);
        }catch (Exception e){
            e.printStackTrace();
        }
        if(searchHits != null){
            // 计算分页总数
            totalPage = (int) ((total - 1) / searchParam.getPageSize() + 1);

            List<SearchHit<SocialAccountCard>> hits = searchHits.getSearchHits();
            if(!CollectionUtils.isEmpty(hits)){
                socialUsers  = hits.stream().map(SearchHit::getContent).collect(Collectors.toList());
                socialUsers.forEach(o->{
                    if(StringUtils.isBlank(o.getPlatform())){
                        o.setPlatform(o.getLinkedinId() != null ? "Linkedin":StringUtils.isNotBlank(o.getLinkedinUsername()) ? "Linkedin": "Facebook");
                    }
                });
            }
            return PageResponse.builder(socialUsers, total, totalPage, searchParam.getOriginPageNo());
        }
        return PageResponse.builderEmpty();
    }

    private NativeSearchQuery getNativeSearchQuery(SocialAccountCardReqVO searchParam, String[] searchFields) {
        NativeSearchQueryBuilder builder = new NativeSearchQueryBuilder();
        if(StringUtils.isNotBlank(searchParam.getSearchKeywords())){
            builder.withQuery(new MultiMatchQueryBuilder(searchParam.getSearchKeywords(), searchFields));
        }
        if(Objects.equals(searchParam.getIsAfricaAccount(), 1)){
            BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();
            boolQueryBuilder.filter(new MatchQueryBuilder("country", AFRICA_COUNTRY_NAMES).operator(Operator.OR));
            builder.withFilter(boolQueryBuilder);
        }
        //es分页从0开始的，而我们的系统是从1开始的，所以-1
        builder.withPageable(PageRequest.of(searchParam.getOriginPageNo() - 1, searchParam.getPageSize()));
        return builder.build();
    }

    private NativeSearchQuery getAfricaNativeSearchQuery(SocialAccountCardReqVO searchParam, String[] searchFields) {
        NativeSearchQueryBuilder builder = new NativeSearchQueryBuilder();
        if(StringUtils.isNotBlank(searchParam.getSearchKeywords())){
            builder.withQuery(new MultiMatchQueryBuilder(searchParam.getSearchKeywords(), searchFields));
        }
        builder.withFilter(new MatchQueryBuilder("country", AVAILABLE_AFRICA_COUNTRY_NAMES.get(searchParam.getOriginPageNo()%AVAILABLE_AFRICA_COUNTRY_NAMES.size())).operator(Operator.AND));
        //es分页从0开始的，而我们的系统是从1开始的，所以-1
        builder.withPageable(PageRequest.of(searchParam.getOriginPageNo() - 1, searchParam.getPageSize()));
        return builder.build();
    }
}
